package com.joymeter.data.service;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.joymeter.common.utils.ConnectionUtils;
import com.joymeter.data.entity.DeviceInfo;
import com.joymeter.data.entity.MeterUser;
import org.springframework.stereotype.Service;

import java.net.InetSocketAddress;
import java.util.List;

/**
 * @ClassName CanalService
 * @Description TODO
 * @Author liang
 * @Date 2018/10/12 17:56
 * @Version 1.0
 **/
@Service
public class CanalService {
    private final static String ip = "192.168.10.168";
    private final static int port = 11111;


    public void connector(){
        // 创建链接
        CanalConnector connector = CanalConnectors.
                newSingleConnector(new InetSocketAddress(ip, port), "example", "", "");
        int batchSize = 1000;
        int emptyCount = 0;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            int totalEmptyCount = 120;
            while (emptyCount < totalEmptyCount) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    emptyCount++;
                    System.out.println("empty count : " + emptyCount);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                    }
                } else {
                    emptyCount = 0;
                    // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
                    printEntry(message.getEntries());
                }

                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }

            System.out.println("empty too many times, exit");
        } finally {
            connector.disconnect();
        }
    }


    private static void printEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            CanalEntry.RowChange rowChage = null;
            try {
                rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }
            String sql = rowChage.getSql();
            System.out.println("============sql"+sql);
            //数据库名
            String databaseName = entry.getHeader().getSchemaName();
            //表名
            String tableName = entry.getHeader().getTableName();

            //过滤数据库，只记录test数据库
            if(!entry.getHeader().getSchemaName().equals("test")) continue;
            if(tableName.equals("meter_user")){
                parseMeterUser(rowChage);
            }else if(tableName.equals("concentrator")){
                parseConcentrator(rowChage);
            }


        }
    }

    /**
     * 解析meter_user表
     */
    private static void parseMeterUser(CanalEntry.RowChange rowChage) {
        //操作类型
        CanalEntry.EventType operationType = rowChage.getEventType();
        //过滤操作类型，只记录，增，改
        if(operationType.toString().equals("DELETE")) return;

        //遍历内容
        for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
            if (operationType == CanalEntry.EventType.INSERT) {
                //插入
                List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                DeviceInfo deviceInfo = new DeviceInfo();
                deviceInfo.meterUserListToDeviceInfo(afterColumnsList);
                System.out.println(deviceInfo.toString());

            }else if (operationType == CanalEntry.EventType.UPDATE){

            }
        }
    }

    /**
     * 解析concentrator表
     */
    private static void parseConcentrator(CanalEntry.RowChange rowChage) {
        //操作类型
        CanalEntry.EventType operationType = rowChage.getEventType();
        if(operationType.toString().equals("DELETE")) return;
        //遍历内容
        for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
            if (operationType == CanalEntry.EventType.INSERT) {
                //插入
                List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                for (CanalEntry.Column column : afterColumnsList) {

                }
            }else if (operationType == CanalEntry.EventType.UPDATE){

            }
        }

    }
}
